package com.ww.flink

import org.apache.flink.api.java.io.TextInputFormat
import org.apache.flink.core.fs.Path
import org.apache.flink.streaming.api.functions.source.FileProcessingMode
import org.apache.flink.streaming.api.scala._


object Flink_try03_monitor_files {
  def main(args: Array[String]): Unit = {
    /**
     * createLocalEnvironment 创建一个本地执行的环境  local
     * createLocalEnvironmentWithWebUI 创建一个本地执行的环境  同时还开启Web UI的查看端口  8081
     * getExecutionEnvironment 根据你执行的环境创建上下文，比如local  cluster
     */
    val env = StreamExecutionEnvironment.getExecutionEnvironment

    //val file: DataStream[String] = env.readTextFile("data/tqdata")
    val filePath = "data/tqdata"
    val format = new TextInputFormat(new Path(filePath))
    val file: DataStream[String] = env.readFile(format, "data/tqdata", FileProcessingMode.PROCESS_CONTINUOUSLY, 10)

    file.print()
    //启动Flink 任务
    env.execute("monitor file flink job")
  }
}
